package com.gmail.dengtao.joe.group.transport.impl.handler;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gmail.dengtao.joe.group.channel.Channel;
import com.gmail.dengtao.joe.group.member.Member;
import com.gmail.dengtao.joe.group.member.MemberUtils;
import com.gmail.dengtao.joe.group.protocol.impl.ACKProtocol;
import com.gmail.dengtao.joe.group.protocol.impl.DATAProtocol;
import com.gmail.dengtao.joe.group.protocol.impl.HELLOProtocol;
import com.gmail.dengtao.joe.group.protocol.impl.IDLEProtocol;
import com.gmail.dengtao.joe.group.transport.impl.AbstractReceiverHandler;
import com.gmail.dengtao.joe.transport.session.IdleStatus;
import com.gmail.dengtao.joe.transport.session.Session;
import com.gmail.dengtao.joe.transport.session.impl.DatagramSession;
import com.gmail.dengtao.joe.transport.session.impl.SocketSession;

/**
 * Default channel receiver handler
 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
 * @version 1.0
 * @since 1.0
 */
public class ChannelReceiverHandler extends AbstractReceiverHandler {
	
	private Logger LOGGER = LoggerFactory.getLogger(this.getClass());
	
	private Member local;
	
	private Map<Session, Member> memberMap = new ConcurrentHashMap<Session, Member>();
	
	public ChannelReceiverHandler(Channel channel, Member local) {
		this.channel = channel;
		this.local = local;
	}

	@Override
	public void sessionCreated(Session session) throws Exception {
		// set idle time for write
		session.setIdleTime(IdleStatus.READ_IDLE, idleTimeForRead);

		Member member = new Member();
		memberMap.put(session, member);
	}

	@Override
	public void sessionOpened(Session session) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[RECEIVER][OPEN][DISCONNECT][MEMBER:UNKOWN][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			member.addEvent(Member.Event.SENDER_CONNECTED);
			
			// Basic information
			InetSocketAddress remoteAddr = null;
			InetSocketAddress localAddr = null;
			if (session instanceof SocketSession) {
				remoteAddr = (InetSocketAddress) ((SocketSession) session).getRemoteSocketAddress();
				localAddr = (InetSocketAddress) ((SocketSession) session).getLocalSocketAddress();
			} else if (session instanceof DatagramSession) {
				remoteAddr = (InetSocketAddress) ((DatagramSession) session).getRemoteSocketAddress();
				localAddr = (InetSocketAddress) ((DatagramSession) session).getLocalSocketAddress();
			}
			
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("[RECEIVER][OPEN][WAITE HELLO...][LOCAL:/" + localAddr.getAddress().getHostAddress() + ":" + localAddr.getPort() + "][REMOTE:/" + remoteAddr.getAddress().getHostAddress() + ":" + remoteAddr.getPort() + "][MEMBER:" + member + "]");
			}
		}
	}

	@Override
	public void sessionIdle(Session session, IdleStatus status) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[RECEIVER][IDLE][DISCONNECT][MEMBER:UNKOWN][IDLE_STATUS:" + status + "][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[RECEIVER][IDLE][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
				// Fire Member Idle!!!!!!!!!!
				interceptorStack.fireMemberIdle(channel, member, com.gmail.dengtao.joe.group.member.IdleStatus.READ_IDLE);
			} else if (member.hasStatus(Member.Status.SENDER_OK)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[RECEIVER][IDLE][WAITE RECEIVER...][SENDER OK][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
			} else {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[RECEIVER][IDLE][CONNECTION IN PROGRESS...][SENDER NOT OK][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
			}
		}
	}

	@Override
	public void dataReceived(Session session, Object data)
			throws Exception {
		Member remote = memberMap.get(session);
		if (remote == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[RECEIVER][RECEIVE][DISCONNECT][MEMBER:UNKOWN][DATA:" + data + "][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			if (data instanceof HELLOProtocol) {	// hello protocol
				HELLOProtocol hello = (HELLOProtocol) data;
				if (remote.hasStatus(Member.Status.SENDER_WAITE_HELLO)) {
					remote.setHost(hello.getHost());
					remote.setPort(hello.getPort());
					remote.setRemoteUuid(hello.getUuid());
	
					if (MemberUtils.getUniqueId(remote, local).equals(hello.getUuid())) {
						String localUuid = MemberUtils.getUniqueId(local, remote);
						Member member = channel.getMember(localUuid);
						if (member == null) {
							remote.setRemoteUuid(hello.getUuid());
							remote.setLocalUuid(localUuid);
							remote.setSender(session);
							remote.addEvent(Member.Event.SENDER_HELLO_RECEIVED);
							if (LOGGER.isDebugEnabled()) {
								LOGGER.debug("[RECEIVER][RECEIVE][WAITE RECEIVER...][SENDER OK][MEMBER:" + remote + "]");
							}
							channel.addMember(remote);
						} else {
							if (!member.hasStatus(Member.Status.SENDER_WAITE_HELLO)) {
								member.addEvent(Member.Event.SENDER_CONNECTED);
							}
							memberMap.put(session, member);
							member.setRemoteUuid(hello.getUuid());
							member.setLocalUuid(localUuid);
							member.setSender(session);
							member.addEvent(Member.Event.SENDER_HELLO_RECEIVED);
							channel.addMember(member);
							if (member.hasStatus(Member.Status.AVAILABLE)) {
								if (LOGGER.isDebugEnabled()) {
									LOGGER.debug("[RECEIVER][SENT][MEMBER ADDED][MEMBER:" + member + "]");
								}
								// Fire Member Add!!!
								interceptorStack.fireMemberAdded(channel, member);
							} else {
								if (LOGGER.isDebugEnabled()) {
									LOGGER.debug("[RECEIVER][RECEIVE][WAITE RECEIVER...][SENDER OK][MEMBER:" + member + "]");
								}
							}
						}
					} else {
						if (LOGGER.isDebugEnabled()) {
							LOGGER.error("[RECEIVER][RECEIVE][DISCONNECT][UUID:ERROR][DATA:" + data + "][MEMBER:" + remote + "]");
						}
						session.close();
						return;
					}
				} else {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.error("[RECEIVER][RECEIVE][DISCONNECT][PROTOCOL:ILLEGAL][DATA:" + data + "][MEMBER:" + remote + "]");
					}
					session.close();
					return;
				}
			} else if (data instanceof IDLEProtocol) {	// idle protocol
				// TODO
				System.err.println("IDLE:" + data);
			} else if (data instanceof DATAProtocol) {	// data protocol
				DATAProtocol protocol = (DATAProtocol) data;
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[RECEIVER][RECEIVE][DATA:" + protocol + "][MEMBER:" + remote + "]");
				}
				// Fire data received!!!
				interceptorStack.fireDataReceived(channel, remote, protocol.getData());
			} else if (data instanceof ACKProtocol) {	// ack protocol
				// TODO
				System.err.println("ACK:" + data);
			} else {	// unkown
				if (LOGGER.isDebugEnabled()) {
					LOGGER.warn("[RECEIVER][RECEIVE][DISCONNECT][PROTOCOL:UNKOWN][DATA:" + data + "][MEMBER:" + remote + "]");
				}
			}
		}
	}

	@Override
	public void sessionClosed(Session session) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[RECEIVER][CLOSE][DISCONNECTED][MEMBER:UNKOWN][SESSION:" + session + "]");
			}
		} else {
			memberMap.remove(session);
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				member.addEvent(Member.Event.SENDER_DISCONNECTED);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[RECEIVER][CLOSE][DISCONNECTED][MEMBER:" + member + "]");
				}
				// Close receiver...
				if (member.getReceiver() != null) member.getReceiver().close();
				
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][MEMBER DISAPPEARED][MEMBER:" + member + "]");
				}
				channel.removeMember(member);
				// Fire Member Disappeared!!!
				interceptorStack.fireMemberDisappeared(channel, member);
			} else {
				if (member.hasStatus(Member.Status.SENDER_WAITE_HELLO) || member.hasStatus(Member.Status.SENDER_OK)) {
					member.addEvent(Member.Event.SENDER_DISCONNECTED);
				}
				// Close receiver...
				if (member.getReceiver() != null) member.getReceiver().close();
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[RECEIVER][CLOSE][DISCONNECTED][MEMBER:" + member + "]");
				}
			}
		}
	}

	@Override
	public void exceptionCaught(Session session, Throwable cause) {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[RECEIVER][EXCEPTION][DISCONNECT][MEMBER:UNKOWN][SESSION:" + session + "]", cause);
			}
			session.close();
		} else {
			if (LOGGER.isDebugEnabled()) {
				if (member.hasStatus(Member.Status.AVAILABLE)) {
					LOGGER.warn("[RECEIVER][EXCEPTION][MEMBER OK][MEMBER:" + member + "]", cause);
				} else if (member.hasStatus(Member.Status.SENDER_OK)) {
					LOGGER.warn("[RECEIVER][EXCEPTION][SENDER OK][MEMBER:" + member + "]", cause);
				} else {
					LOGGER.warn("[RECEIVER][EXCEPTION][SENDER NOT OK][MEMBER:" + member + "]", cause);
				}
			}
			// Fire Exception Caught!!
			interceptorStack.fireExceptionCaught(channel, member, cause);
		}
	}

}